/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.handler.codec;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.ChannelInputShutdownEvent;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.StringUtil;

import java.util.List;

import static io.netty.util.internal.ObjectUtil.checkPositive;
import static java.lang.Integer.MAX_VALUE;

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {

    public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
            if (!cumulation.isReadable() && in.isContiguous()) {

                cumulation.release();
                return in;
            }
            try {
                //获取从socket加载的bytebuf可读数据量，
                final int required = in.readableBytes();
                //条件成立 ，说明堆积区可用空间不足以存放本次读取到的数据
                if (required > cumulation.maxWritableBytes() ||
                        (required > cumulation.maxFastWritableBytes() && cumulation.refCnt() > 1) ||
                        cumulation.isReadOnly()) {
                    //扩容 ， 内存池 ， 堆积区 ， socket获取到的数据
                    return expandCumulation(alloc, cumulation, in);
                }
                //将msg尾插到堆积区
                cumulation.writeBytes(in, in.readerIndex(), required);
                //设置读索引
                in.readerIndex(in.writerIndex());
                //返回堆积区
                return cumulation;
            } finally {
                //内存释放
                in.release();
            }
        }
    };

    public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
            if (!cumulation.isReadable()) {
                cumulation.release();
                return in;
            }
            CompositeByteBuf composite = null;
            try {
                if (cumulation instanceof CompositeByteBuf && cumulation.refCnt() == 1) {
                    composite = (CompositeByteBuf) cumulation;
                    // Writer index must equal capacity if we are going to "write"
                    // new components to the end
                    if (composite.writerIndex() != composite.capacity()) {
                        composite.capacity(composite.writerIndex());
                    }
                } else {
                    composite = alloc.compositeBuffer(Integer.MAX_VALUE).addFlattenedComponents(true, cumulation);
                }
                composite.addFlattenedComponents(true, in);
                in = null;
                return composite;
            } finally {
                if (in != null) {
                    // We must release if the ownership was not transferred as otherwise it may produce a leak
                    in.release();
                    // Also release any new buffer allocated if we're not returning it
                    if (composite != null && composite != cumulation) {
                        composite.release();
                    }
                }
            }
        }
    };
    //初始状态。
    private static final byte STATE_INIT = 0;
    //调用子类实现的decode方法时期状态。
    private static final byte STATE_CALLING_CHILD_DECODE = 1;
    //当decoder从pipeline移除出去之后，状态会修改为removed_pending状态。
    private static final byte STATE_HANDLER_REMOVED_PENDING = 2;
    //堆积区
    ByteBuf cumulation;
    //小块数据进来以后，他会负责将数据合并到堆积区内。
    private Cumulator cumulator = MERGE_CUMULATOR;
    //默认是false，如果是true，每次channelRRead只解码一个frame。
    private boolean singleDecode;
    // false，代表   cumulation 为 null，true代表不为空。
    //当堆积区的数据处理完以后，会将堆积区释放，然后将内存归还
    private boolean first;

    private boolean firedChannelRead;

    private byte decodeState = STATE_INIT;
    //清理堆积区已读空间的阈值，当堆积区的readIndex比较大的时候，堆积区可写的空间会变小，通过整理的方法可以调整堆积区的读指针和写指针，复用空闲的内存。
    private int discardAfterReads = 16;
    //已经读了多少次。
    private int numReads;

    protected ByteToMessageDecoder() {
        ensureNotSharable();
    }

    public void setSingleDecode(boolean singleDecode) {
        this.singleDecode = singleDecode;
    }

    public boolean isSingleDecode() {
        return singleDecode;
    }

    public void setCumulator(Cumulator cumulator) {
        this.cumulator = ObjectUtil.checkNotNull(cumulator, "cumulator");
    }

    public void setDiscardAfterReads(int discardAfterReads) {
        checkPositive(discardAfterReads, "discardAfterReads");
        this.discardAfterReads = discardAfterReads;
    }

    protected int actualReadableBytes() {
        return internalBuffer().readableBytes();
    }

    protected ByteBuf internalBuffer() {
        if (cumulation != null) {
            return cumulation;
        } else {
            return Unpooled.EMPTY_BUFFER;
        }
    }

    @Override
    public final void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        if (decodeState == STATE_CALLING_CHILD_DECODE) {
            decodeState = STATE_HANDLER_REMOVED_PENDING;
            return;
        }
        ByteBuf buf = cumulation;
        if (buf != null) {
            // Directly set this to null so we are sure we not access it in any other method here anymore.
            cumulation = null;
            numReads = 0;
            int readable = buf.readableBytes();
            if (readable > 0) {
                ctx.fireChannelRead(buf);
                ctx.fireChannelReadComplete();
            } else {
                buf.release();
            }
        }
        handlerRemoved0(ctx);
    }

    protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //msg是channel.unsafe到channel底层关联的socket读取数据的时候使用的bytebuf。
        if (msg instanceof ByteBuf) {
            //可以简单的看成一个list，一个池化的list，空间换时间的思想。
            CodecOutputList out = CodecOutputList.newInstance();
            try {
                first = cumulation == null;
                //将读取的msg合并到堆积区，【尾插】
                //参数一：内存池
                //参数二：堆积区
                //参数三：msg
                cumulation = cumulator.cumulate(ctx.alloc(),
                        first ? Unpooled.EMPTY_BUFFER : cumulation, (ByteBuf) msg);
                //参数一：ctx
                //参数二：堆积区
                // 参数三：已经解码的数据集合
                //数据解码逻辑
                callDecode(ctx, cumulation, out);
                /*
                1. 堆积区的数据全部解析成 frame ， 并且已经发送到 decoder 后面的 handler 去处理了（out可能还有剩余未发送的数据）
                2. 堆积区内的数据是半包数据
                 */
            } catch (DecoderException e) {
                throw e;
            } catch (Exception e) {
                throw new DecoderException(e);
            } finally {
                //条件成立：说明 堆积区内的数据全部处理完了
                if (cumulation != null && !cumulation.isReadable()) {
                    numReads = 0;
                    //将堆积区占用的内存释放
                    cumulation.release();
                    //堆积区字段设置为null，下次再有 channelRead 事件的时候重建堆积区
                    cumulation = null;
                } else if (++ numReads >= discardAfterReads) {
                    //什么时候执行到else ， 堆积区内剩余的数据是半包数据...
                    //++ numReads ，记录堆积区累加过的数据次数
                    numReads = 0;
                    //调整堆积区 ByteBuf 读指针和写指针，重复利用空闲空间
                    discardSomeReadBytes();
                }

                //如果out里面还有数据，获取一下，传到下一个 ctx ，
                int size = out.size();
                firedChannelRead |= out.insertSinceRecycled();
                fireChannelRead(ctx, out, size);
                //回收 集合
                out.recycle();
            }
        } else {
            ctx.fireChannelRead(msg);
        }
    }

    static void fireChannelRead(ChannelHandlerContext ctx, List<Object> msgs, int numElements) {
        if (msgs instanceof CodecOutputList) {
            fireChannelRead(ctx, (CodecOutputList) msgs, numElements);
        } else {
            for (int i = 0; i < numElements; i++) {
                ctx.fireChannelRead(msgs.get(i));
            }
        }
    }

    static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
        for (int i = 0; i < numElements; i ++) {
            ctx.fireChannelRead(msgs.getUnsafe(i));
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        numReads = 0;
        discardSomeReadBytes();
        if (!firedChannelRead && !ctx.channel().config().isAutoRead()) {
            ctx.read();
        }
        firedChannelRead = false;
        ctx.fireChannelReadComplete();
    }

    protected final void discardSomeReadBytes() {
        if (cumulation != null && !first && cumulation.refCnt() == 1) {
            // discard some bytes if possible to make more room in the
            // buffer but only if the refCnt == 1  as otherwise the user may have
            // used slice().retain() or duplicate().retain().
            //
            // See:
            // - https://github.com/netty/netty/issues/2327
            // - https://github.com/netty/netty/issues/1764
            cumulation.discardSomeReadBytes();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        channelInputClosed(ctx, true);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof ChannelInputShutdownEvent) {
            // The decodeLast method is invoked when a channelInactive event is encountered.
            // This method is responsible for ending requests in some situations and must be called
            // when the input has been shutdown.
            channelInputClosed(ctx, false);
        }
        super.userEventTriggered(ctx, evt);
    }

    private void channelInputClosed(ChannelHandlerContext ctx, boolean callChannelInactive) {
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            channelInputClosed(ctx, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Exception e) {
            throw new DecoderException(e);
        } finally {
            try {
                if (cumulation != null) {
                    cumulation.release();
                    cumulation = null;
                }
                int size = out.size();
                fireChannelRead(ctx, out, size);
                if (size > 0) {
                    // Something was read, call fireChannelReadComplete()
                    ctx.fireChannelReadComplete();
                }
                if (callChannelInactive) {
                    ctx.fireChannelInactive();
                }
            } finally {
                // Recycle in all cases
                out.recycle();
            }
        }
    }

    void channelInputClosed(ChannelHandlerContext ctx, List<Object> out) throws Exception {
        if (cumulation != null) {
            callDecode(ctx, cumulation, out);
            decodeLast(ctx, cumulation, out);
        } else {
            decodeLast(ctx, Unpooled.EMPTY_BUFFER, out);
        }
    }
    //参数一：ctx
    //参数二：堆积区
    //参数三：已经解码的数据集合 ，默认此时是空
    protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        try {
            //堆积区已经没有待处理的数据了
            while (in.isReadable()) {
                //第一次进来是 0
                int outSize = out.size();

                if (outSize > 0) {
                    fireChannelRead(ctx, out, outSize);
                    out.clear();

                    // Check if this handler was removed before continuing with decoding.
                    // If it was removed, it is not safe to continue to operate on the buffer.
                    //
                    // See:
                    // - https://github.com/netty/netty/issues/4635
                    if (ctx.isRemoved()) {
                        break;
                    }
                    outSize = 0;
                }

                //获取待处理长度大小
                int oldInputLength = in.readableBytes();
                //内部调用了decode方法，这个方法是抽象的，由子类实现，子类将in的数据解析成业务协议的frame，并且放入到out。
                decodeRemovalReentryProtection(ctx, in, out);

                if (ctx.isRemoved()) {
                    break;
                }

                //一般条件成立， 0  ==  0 ，也就是啥都没解析出来
                //1. 当前要解析的数据属于半包数据
                //2. 数据过长，超出解码器解析限制，数据被丢弃
                if (outSize == out.size()) {
                    if (oldInputLength == in.readableBytes()) {
                        break;
                    } else {
                        continue;
                    }
                }

                //解析前后数据大小，抛异常，，，
                //执行到这里已经排除了半包，说明解析出来了数据，但是堆积区的大小没变，一定是子类实现的逻辑有问题
                if (oldInputLength == in.readableBytes()) {
                    throw new DecoderException(
                            StringUtil.simpleClassName(getClass()) +
                                    ".decode() did not read anything but decoded a message.");
                }
                //如果成立，每次只能解析一个，跳出循环。
                if (isSingleDecode()) {
                    break;
                }
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Exception cause) {
            throw new DecoderException(cause);
        }
    }

    protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

    final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
            throws Exception {
        decodeState = STATE_CALLING_CHILD_DECODE;
        try {
            decode(ctx, in, out);
        } finally {
            boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
            decodeState = STATE_INIT;
            if (removePending) {
                fireChannelRead(ctx, out, out.size());
                out.clear();
                handlerRemoved(ctx);
            }
        }
    }

    protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.isReadable()) {
            // Only call decode() if there is something left in the buffer to decode.
            // See https://github.com/netty/netty/issues/4386
            decodeRemovalReentryProtection(ctx, in, out);
        }
    }
    //内存池 ， 堆积区 ， socket获取到的数据
    static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf oldCumulation, ByteBuf in) {
        //原堆积区未处理数据量大小
        int oldBytes = oldCumulation.readableBytes();
        //socket获取到的数据量大小
        int newBytes = in.readableBytes();
        //计算上面二者的总大小
        int totalBytes = oldBytes + newBytes;
        //申请内存
        //calculateNewCapacity 计算分配的内存大小
        //1. 如果totalBytes==4m，直接返回
        //2.如果totalBytes>4m,判断如果是4m的整数倍就返回整数倍，否则凑成整数倍返回
        //3.如果totalBytes<4m,从64开始不停的<< 1 ,直到大于totalBytes
        ByteBuf newCumulation = alloc.buffer(alloc.calculateNewCapacity(totalBytes, MAX_VALUE));
        //待释放内存，这里为什么设置成新申请的内存呢？
        //目测是为了防止下面的逻辑发生异常，新申请的内存没有来得及释放
        ByteBuf toRelease = newCumulation;
        try {
            // 将原堆积区数据全部导入到新的堆积区
            newCumulation.setBytes(0, oldCumulation, oldCumulation.readerIndex(), oldBytes)
                    //将从socket获取的数据也写入到堆积区
                .setBytes(oldBytes, in, in.readerIndex(), newBytes)
                    //更新写索引
                .writerIndex(totalBytes);
            //更新读索引
            in.readerIndex(in.writerIndex());
            //设置待释放内存
            toRelease = oldCumulation;
            return newCumulation;
        } finally {
            //内存释放逻辑
            toRelease.release();
        }
    }

    public interface Cumulator {
        ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
    }
}
